package com.zyy.joseata.mq;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.ReturnListener;
import com.zyy.joseata.bo.MqMsgBO;
import com.zyy.joseata.constant.TxEventTypeConst;
import com.zyy.joseata.socket.SocketClient;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author zhangyy
 */
@Slf4j
public class JoSeataMqAck implements ConfirmListener, ReturnListener {

    private SocketClient socketClient;

    public JoSeataMqAck(SocketClient socketClient){
        this.socketClient = socketClient;
    }

    public static ConcurrentHashMap<Long, MqMsgBO> deliveryTagKeyMsgMap = new ConcurrentHashMap<>();

    public static AtomicLong curDeliverTag = null;

    @SneakyThrows
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        if(curDeliverTag == null){
            curDeliverTag = new AtomicLong(deliveryTag);
            for(MqMsgBO mqMsg:deliveryTagKeyMsgMap.values()){
                socketClient.send(TxEventTypeConst.MQ_TX_SEND_SUCCESS, JSONObject.toJSONString(mqMsg));
            }
        }else{
            for(long i=curDeliverTag.longValue();i<deliveryTag;i++){
                MqMsgBO mqMsg = deliveryTagKeyMsgMap.get(i);
                socketClient.send(TxEventTypeConst.MQ_TX_SEND_SUCCESS, JSONObject.toJSONString(mqMsg));
            }
        }
    }

    @SneakyThrows
    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        if(curDeliverTag == null){
            curDeliverTag = new AtomicLong(deliveryTag);
            for(MqMsgBO mqMsg:deliveryTagKeyMsgMap.values()){
                socketClient.send(TxEventTypeConst.MQ_TX_SEND_FAIL, JSONObject.toJSONString(mqMsg));
            }
        }else{
            for(long i=curDeliverTag.longValue();i<deliveryTag;i++){
                MqMsgBO mqMsg = deliveryTagKeyMsgMap.get(i);
                socketClient.send(TxEventTypeConst.MQ_TX_SEND_FAIL, JSONObject.toJSONString(mqMsg));
            }
        }
    }

    @SneakyThrows
    @Override
    public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties basicProperties, byte[] bytes) {
        MqMsgBO mqMsg = new MqMsgBO(exchange,routingKey,basicProperties.getMessageId(),JSONObject.parseObject(bytes, Map.class));
        log.info("消息路由失败! 消息:{}",mqMsg);
        socketClient.send(TxEventTypeConst.MQ_TX_SEND_FAIL,JSONObject.toJSONString(mqMsg));
    }
}
